package com.sh.data.engine.domain.shims.mqtt.util;

import com.sh.data.engine.domain.shims.kafka.util.SocketUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.RandomStringUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.tuple.Pair;
import org.eclipse.paho.client.mqttv3.*;
import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence;

import java.io.IOException;
import java.util.Arrays;
import java.util.List;
import java.util.concurrent.TimeUnit;

/**
 * @author: mengzheng.mhc
 * @date: 2024/6/11 16:01
 */
@Slf4j
public class MQTTUtil {

    private static final String ARG_ERROR_MSG = "mqtt server配置有误，请检查";
    private static final int DEFAULT_QOS = 1;

    public static void testConnection(String address, Integer timeoutInSeconds)
        throws IOException, IllegalArgumentException {
        if (timeoutInSeconds == null || timeoutInSeconds < 0) {
            timeoutInSeconds = 10;
        }

        address = StringUtils.lowerCase(address);
        address = StringUtils.removeAll(address, "tcp://");
        address = StringUtils.removeAll(address, "ssl://");
        try {
            List<Pair<String, Integer>> list = SocketUtil.parseHostAndPort(address, 1883);
            SocketUtil.testSocketConnection(list, timeoutInSeconds);
        } catch (IllegalArgumentException e) {
            throw new IllegalArgumentException(ARG_ERROR_MSG, e);
        }
    }

    public static List<String> preview(String host, String topic) throws Exception {
        IMqttClient mqttClient = null;
        try {
            mqttClient = createMqttClient(host);
            MqttConnectOptions options = createMqttConnectOptions();
            mqttClient.connect(options);
            var listener = new DataIntegrationEngineMqttMessageListener();
            mqttClient.subscribe(topic, DEFAULT_QOS, listener);
            String payload = listener.getLastMessage(30);
            if (payload == null) {
                throw new RuntimeException("超时，未获取到消息，请检查mqtt server是否收到消息");
            }
            return Arrays.asList(listener.getLastMessage(30));
        } catch (RuntimeException e) {
            throw e;
        } catch (MqttException e) {
            throw new RuntimeException("MQTT客户端连接服务器失败，请检查配置:" + host + " ,topic:" + topic);
        } catch (Exception e) {
            throw e;
        } finally {
            if (mqttClient != null) {
                mqttClient.disconnect();
            }
        }
    }

    public static IMqttClient createMqttClient(String host)
        throws IllegalArgumentException, MqttException {
        host = StringUtils.trim(host);
        checkHost(host);
        String clientId = getRandomClientId();
        // 使用内存做数据持久化
        return new MqttClient(host, clientId, new MemoryPersistence());
    }

    private static class DataIntegrationEngineMqttMessageListener implements IMqttMessageListener {
        private volatile String lastMessage;

        public String getLastMessage(Integer timeoutInSeconds) {
            if (timeoutInSeconds == null || timeoutInSeconds < 0) {
                timeoutInSeconds = 10;
            }
            int timeoutInMills = 1000 * timeoutInSeconds;
            long start = System.currentTimeMillis();
            while (true) {
                if (StringUtils.isNotBlank(lastMessage)) {
                    return lastMessage;
                }
                if (System.currentTimeMillis() - start >= timeoutInMills) {
                    return lastMessage;
                }
                try {
                    TimeUnit.SECONDS.sleep(2);
                } catch (InterruptedException ignore) {
                }
            }
        }

        @Override
        public void messageArrived(String topic, MqttMessage mqttMessage) throws Exception {
            lastMessage = new String(mqttMessage.getPayload(), "utf-8");
        }
    }

    public static MqttConnectOptions createMqttConnectOptions() {
        return createMqttConnectOptions(null, null, null);
    }

    public static MqttConnectOptions createMqttConnectOptions(
        String username, String password, Integer timeoutInSeconds) {
        if (username == null) {
            username = "admin";
        }
        if (password == null) {
            password = "admin";
        }
        if (timeoutInSeconds == null || timeoutInSeconds < 0) {
            timeoutInSeconds = 10;
        }
        MqttConnectOptions options = new MqttConnectOptions();
        options.setUserName(username);
        options.setPassword(password.toCharArray());
        // 自动重连
        options.setAutomaticReconnect(true);
        // 是否临时会话
        options.setCleanSession(true);
        // 超时时间,单位是秒
        options.setConnectionTimeout(timeoutInSeconds);
        return options;
    }

    private static void checkHost(String host) {
        if (!StringUtils.startsWithIgnoreCase(host, "tcp://")
            && !StringUtils.startsWithIgnoreCase(host, "ssl://")) {
            throw new IllegalArgumentException("mqtt server配置有误，必须以tcp://或ssl://开头");
        }
    }

    // MQTT clientId长度有限制，为兼容不同版本，最好长度在23个字符以内
    // https://blog.csdn.net/tiantiandjava/article/details/44154047
    private static String getRandomClientId(String prefix, int length) {
        return prefix + RandomStringUtils.randomAlphanumeric(length);
    }

    public static String getRandomClientId() {
        return "data_engine_mqtt_" + RandomStringUtils.randomAlphanumeric(10);
    }

  /*  public static List<String> preview(String host, String topic) throws Exception {
    ArrayList<String> data = Lists.newArrayList();
    MQTT mqtt = new MQTT();
    Snowflake snowflake = IdUtil.getSnowflake();

    mqtt.setHost(host);
    mqtt.setClientId(snowflake.nextIdStr());
    mqtt.setCleanSession(false);

    FutureConnection futureConnection = mqtt.futureConnection();

    futureConnection.connect().await();

    Future<byte[]> subscribe =
        futureConnection.subscribe(new Topic[] {new Topic(topic, QoS.EXACTLY_ONCE)});

    subscribe.await();

    Future<Message> receive = futureConnection.receive();
    Message message = null;

    try {
      message = receive.await(30, TimeUnit.SECONDS);
    } catch (Exception e) {
      throw new RuntimeException("超时未获取到数据");
    }
    if (null != message) {
      data.add(new String(message.getPayload()));
      message.ack();
    }

    return data;
  } */
}
